package org.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HBaseDML {

    public static Connection connection = HBaseConnection.connection;

    /**
     * 插入数据
     * @param namespace 命名空间
     * @param tableName 表格名称
     * @param rowKey rowKey
     * @param columnFamily 列族名称
     * @param columnName 列名
     * @param value 值
     */
    public static void putCell(
            String namespace,
            String tableName,
            String rowKey,
            String columnFamily,
            String columnName,
            String value
    ) throws IOException {
        Table table = getTable(namespace, tableName);
        if(null==table){
            return;
        }
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily),
                Bytes.toBytes(columnName),Bytes.toBytes(value));
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
    }

    /**
     * 读取数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey rowKey
     * @param columnFamily 列族
     * @param columnName    列名
     */
    public static Cell[] getCells(
            String namespace,
            String tableName,
            String rowKey,
            String columnFamily,
            String columnName
    ) throws IOException {
        Table table = getTable(namespace, tableName);

        // 创建get对象
        Get get = new Get(Bytes.toBytes(rowKey));
        //某一列数据
        get.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));
        // 也可以添加其他参数

        //设置读取数据的版本，作者会提供一个默认的版本，例如你只写了1，但是底层还是会帮你
        // 维护10个版本
        get.readAllVersions();
        // 如果不添加参数就是直接读取rowkey这一整行的数据
        Cell[] cells = null;
        try {
            Result result = table.get(get);
            cells = result.rawCells();
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
        return cells;
    }

    /**
     * 扫描数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param startRow 开始的row 包含
     * @param stopRow  结束的row 不包含
     */
    public static void scanRows(String namespace,
                                String tableName,
                                String startRow,
                                String stopRow) throws IOException {
        Table table = getTable(namespace, tableName);
        // 创建scan对象
        Scan scan = new Scan();
        // 不填写参数，将会全表扫描
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));
        // 读取结果
        ResultScanner scanner = null;
        try {
            scanner = table.getScanner(scan);
            // Result 记录 一行的Cell，而scanner记录多个Result，二维数组
            for (Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    System.out.printf("%s - %s -%s -%s \t",
                            new String(CellUtil.cloneValue(cell)),
                            new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)),
                            new String(CellUtil.cloneRow(cell)));
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
    }

    /**
     *  扫描过滤数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param startRow 开始的row 包含
     * @param stopRow  结束的row 不包含
     * @param columnsFamily 列族
     * @param columnName 列名
     * @param value value值
     * @throws IOException
     */
    public static void filterRows(String namespace,
                                String tableName,
                                String startRow,
                                String stopRow,
                                  String columnsFamily,
                                  String columnName,
                                  String value) throws IOException {
        Table table = getTable(namespace, tableName);
        // 创建scan对象
        Scan scan = new Scan();
        // 不填写参数，将会全表扫描
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));

        //添加多个过滤
        FilterList filterList = new FilterList();

        /*
        创建过滤器，HBase中有两种过滤器
        1.结果值保留当前列的数据
         */
        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
                // 列族
                Bytes.toBytes(columnsFamily),
                // 列名
                Bytes.toBytes(columnName),
                // 比较关系，这是一个枚举
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
        );

        // 2.结果保留整行
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                // 列族
                Bytes.toBytes(columnsFamily),
                // 列名
                Bytes.toBytes(columnName),
                // 比较关系，这是一个枚举
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
        );
        filterList.addFilter(singleColumnValueFilter);
//        filterList.addFilter(columnValueFilter);
        // 添加过滤
        scan.setFilter(filterList);

        /*
        * 两者的区别在于，
        * ColumnValueFilter 会准备找到符合目标的那一个单元格
        *SingleColumnValueFilter 满足目标的第一个单元格除此之外，还会保留
        * 没有当前列的数据
        *
        * 意味着，如果多个rowkey，有的rowkey不含有你当前的过滤条件，会给你保留下来
        * 例如我要查找name为ming的
        * 666 - info1 -age -1002 	ming - info1 -name -1002 对于已经符合的条件，整个
        * rowkey我也给你保留
            666 - info1 -age -1004 rowkey为1004的没有name，所以我依旧给你保留了
        * */

        try {
            ResultScanner scanner = table.getScanner(scan);
            // Result 记录 一行的Cell，而scanner记录多个Result，二维数组
            for (Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    System.out.printf("%s - %s -%s -%s \t",
                            new String(CellUtil.cloneValue(cell)),
                            new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)),
                            new String(CellUtil.cloneRow(cell)));
                }
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
    }

    /**
     * 删除一行的数据
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey    rowKey
     * @param columnsFamily 列族
     * @param columnName    列名
     */
    public static void deleteColumn(
            String namespace,
            String tableName,
            String rowKey,
            String columnsFamily,
            String columnName
    ) throws IOException {
        Table table = getTable(namespace, tableName);

        Delete delete = new Delete(Bytes.toBytes(rowKey));
        // 添加列信息
        // 删除一个版本 相当于delete
//        delete.addColumn(Bytes.toBytes(columnsFamily),
//                Bytes.toBytes(columnName));
        // 删除多个版本 相当于deleteall
        delete.addColumns(
                Bytes.toBytes(columnsFamily),
                Bytes.toBytes(columnName)
        );
        try {
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
    }

    private static Table getTable(
            String namespace,
            String tableName
    ) throws IOException {
        if (!HBaseDDL.isTableExists(namespace, tableName)){
            System.out.println("表不存在");
            return null;
        }
        return connection.getTable(TableName.valueOf(namespace, tableName));
    }

    /*
    * 统计hbase条数，资料来源【暂时还未在环境上验证】
    * https://blog.csdn.net/m0_52735414/article/details/128046428
    * 查询效率由低到高
    *  hbase> count 'ns1:t1'
         hbase> count 't1'
         hbase> count 't1', INTERVAL => 100000
         hbase> count 't1', CACHE => 1000
         hbase> count 't1', INTERVAL => 10, CACHE => 1000
    * */

    /**
     * 使用 FirstKeyOnlyFilter过滤器的scan进行全表扫描
     * @param tablename
     */
    public void rowCountByScanFilter(String tablename){
        long rowCount = 0;
        try {
            //计时
//            StopWatch stopWatch = new StopWatch();
//            stopWatch.start();

            TableName name=TableName.valueOf(tablename);
            //connection为类静态变量
            Table table = connection.getTable(name);
            Scan scan = new Scan();
            //FirstKeyOnlyFilter只会取得每行数据的第一个kv，提高count速度
            scan.setFilter(new FirstKeyOnlyFilter());

            ResultScanner rs = table.getScanner(scan);
            for (Result result : rs) {
                rowCount += result.size();
            }

//            stopWatch.stop();
            System.out.println("RowCount: " + rowCount);
//            System.out.println("统计耗时：" +stopWatch.toString());
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    /*使用hbase自带的工具查询数量
    *这种方式效率非常高！利用了hbase jar中自带的统计行数的工具类

       通过 $HBASE_HOME/bin/hbase 命令执行：

    # hbase org.apache.hadoop.hbase.mapreduce.RowCounter '你的表名'
    * */

    /*
    * 与hive结合，创建hive的关联表，记录条数，之后，直接使用hivesql查询hive表来进行统计
    *
    *
        CREATE EXTERNAL TABLE Test(
        mRID string,
        name string,
        nominalVoltage FLOAT
        )
        * --行的格式
        ROW FORMAT SERDE'org.apache.hadoop.hive.hbase.HBaseSerDe'
        STORED BY'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
        * -- 添加指定的序列化算法
        WITH SERDEPROPERTIES("hbase.columns.mapping" = ":key,pw:name,pw:nominalVoltage")
        * -- 额外属性
        TBLPROPERTIES("hbase.table.name"= "test");
        *
        * 直接统计
select   count(0)  from   test ;
    * */

    /*
    * 使用携程，目前hbase最快的方式
    *Table注册了Coprocessor之后，在执行AggregationClient的时候，会将RowCount分散到Table的每一个Region上，Region内RowCount的计算，是通过RPC执行调用接口，由Region对应的RegionServer执行InternalScanner进行的。

因此，性能的提升有两点原因:

1.分布式统计。将原来客户端按照Rowkey的范围单点进行扫描，然后统计的方式，换成了由所有Region所在RegionServer同时计算的过程。

2.使用了在RegionServer内部执行使用了InternalScanner。这是距离实际存储最近的Scanner接口，存取更加快捷。
    * */
    public void rowCountByCoprocessor(String tablename){
        try {
            //提前创建connection和conf
            Configuration conf = HBaseConnection.configuration();

            Admin admin = connection.getAdmin();
            TableName name=TableName.valueOf(tablename);
            //先disable表，添加协处理器后再enable表
            admin.disableTable(name);
            HTableDescriptor descriptor = admin.getTableDescriptor(name);
            String coprocessorClass = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
            if (! descriptor.hasCoprocessor(coprocessorClass)) {
                descriptor.addCoprocessor(coprocessorClass);
            }
            admin.modifyTable(name, descriptor);
            admin.enableTable(name);

            //计时
//            StopWatch stopWatch = new StopWatch();
//            stopWatch.start();
            Scan scan = new Scan();
            AggregationClient aggregationClient = new AggregationClient(conf);

            System.out.println("RowCount: " + aggregationClient.rowCount(name, new LongColumnInterpreter(), scan));
//            stopWatch.stop();
//            System.out.println("统计耗时：" +stopWatch.toString());
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
//        putCell("default","stu",
//               "2001","info1","name",
//                "pop");

        // 读取数据小案例
//        Cell[] cells = getCells("default", "stu",
//                "2001", "info1", "name");
//        for (Cell cell : cells) {
//            // 如果直接读取cell会乱码，使用作者提供工具不会
//            String s = new String(CellUtil.cloneValue(cell));
//            System.out.println(s);
//        }

//        scanRows("default", "stu",
//                "1001","2001");

//        filterRows("default", "stu",
//                "1001","2001",
//                "info1","name","ming");

        deleteColumn("default", "stu",
                "2001","info1","name");
        HBaseConnection.close();
    }
}
